package io.kiki.sba.registry.server.metadata.monitor.impl;

import io.kiki.sba.registry.common.model.metaserver.nodes.DataStoreNode;
import io.kiki.sba.registry.common.model.slot.*;
import io.kiki.sba.registry.exception.InitializeException;
import io.kiki.sba.registry.lifecycle.impl.AbstractLifecycle;
import io.kiki.sba.registry.server.metadata.bootstrap.config.ServerConfig;
import io.kiki.sba.registry.server.metadata.monitor.Metrics;
import io.kiki.sba.registry.server.metadata.monitor.SlotStats;
import io.kiki.sba.registry.server.metadata.monitor.SlotsStats;
import io.kiki.sba.registry.server.metadata.slot.SlotManager;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class SlotsStatsImpl extends AbstractLifecycle implements SlotsStats {
    private static final Logger logger = LoggerFactory.getLogger(SlotsStatsImpl.class);
    private final SlotManager slotManager;

    private final Map<Integer, SlotStats> slotStatses = new ConcurrentHashMap<>();

    private final ReadWriteLock lock = new ReentrantReadWriteLock();

    private final ServerConfig serverConfig;

    public SlotsStatsImpl(SlotManager slotManager, ServerConfig serverConfig) {
        this.slotManager = slotManager;
        this.serverConfig = serverConfig;
    }

    @Override
    protected void doInitialize() throws InitializeException {
        super.doInitialize();
        for (int slotId = 0; slotId < SlotConfig.SLOT_NUM; slotId++) {
            Slot slot = slotManager.getSlots().getSlot(slotId);
            if (slot == null) {
                slot = new Slot(slotId, null, 0L, Collections.emptyList());
            }
            slotStatses.put(slotId, new SlotStatsImpl(slot, serverConfig.getDataReplicateMaxGapMillis()));
        }
    }

    @Override
    public boolean isSlotLeadersStable() {
        lock.readLock().lock();
        try {
            if (slotManager.getSlots() == Slots.INIT) {
                logger.warn("[isSlotLeadersStable] slot table empty now");
                return false;
            }
            for (int slotId = 0; slotId < SlotConfig.SLOT_NUM; slotId++) {
                Slot slot = slotManager.getSlots().getSlot(slotId);
                if (slot == null) {
                    logger.error("[isSlotLeadersStable] slot manager has no slot: [{}]", slotId);
                    return false;
                }
                String leader = slot.getLeaderDataStoreNodeId();
                SlotStats slotStats = slotStatses.get(slotId);
                if (StringUtils.isBlank(leader) || !slotStats.getSlot().getLeaderDataStoreNodeId().equals(leader) || !slotStatses.get(slotId).isLeaderStable()) {
                    logger.warn("[isSlotLeadersStable]slot[{}] leader[{}] not stable", slotId, leader);
                    return false;
                }
            }
            return true;
        } finally {
            lock.readLock().unlock();
        }
    }

    @Override
    public boolean isSlotFollowersStable() {
        lock.readLock().lock();
        try {
            if (slotManager.getSlots() == Slots.INIT) {
                logger.warn("[isSlotFollowersStable] slot table empty now");
                return false;
            }
            for (Map.Entry<Integer, SlotStats> entry : slotStatses.entrySet()) {
                Slot slot = slotManager.getSlots().getSlot(entry.getKey());
                if (slot == null) {
                    logger.error("[isSlotFollowersStable] slot manager has no slot: [{}]", entry.getKey());
                    return false;
                }
                Set<String> followers = slot.getFollowerDataStoreNodeIdSet();
                for (String follower : followers) {
                    if (!entry.getValue().isFollowerStable(follower)) {
                        logger.warn("[isSlotFollowersStable]slot[{}] follower not stable {}", entry.getKey(), entry.getValue());
                        return false;
                    }
                }
            }
            return true;
        } finally {
            lock.readLock().unlock();
        }
    }

    @Override
    public void checkSlotStatuses(DataStoreNode dataStoreNode, List<BaseSlotStatus> slotStatusList) {
        try {
            lock.writeLock().lock();
            for (BaseSlotStatus slotStatus : slotStatusList) {
                int slotId = slotStatus.getSlotId();
                SlotStats slotStats = slotStatses.get(slotId);
                if (slotStats == null || slotStats.getSlot() == null) {
                    continue;
                }
                if (slotStats.getSlot().getLeaderEpoch() > slotStatus.getSlotLeaderEpoch()) {
                    logger.warn("[checkSlotStatuses] won't update slot status, slot[{}] leader-epoch[{}] is less than current[{}]", slotId, slotStatus.getSlotLeaderEpoch(), slotStats.getSlot().getLeaderEpoch());
                    continue;
                } else
                    if (slotStats.getSlot().getLeaderEpoch() < slotStatus.getSlotLeaderEpoch()) {
                        Metrics.DataSlot.setDataSlotGreaterThanMeta(dataStoreNode.getIp(), slotId);
                        logger.error("[checkSlotStatuses] won't update slot status, slot[{}] leader-epoch[{}] reported by data({}) is more than current[{}]", slotId, slotStatus.getSlotLeaderEpoch(), dataStoreNode.getIp(), slotStats.getSlot().getLeaderEpoch());
                        continue;
                    }
                if (!slotStats.getSlot().equals(slotManager.getSlots().getSlot(slotId))) {
                    logger.error("[checkSlotStatuses] slot reported by data({}) is not equals with mine({}), not update", slotStats.getSlot(), slotManager.getSlots().getSlot(slotId));
                    continue;
                }

                if (slotStatus.getSlotRole() == SlotRole.leader) {
                    if (!slotStats.getSlot().getLeaderDataStoreNodeId().equals(dataStoreNode.getIp())) {
                        logger.error("[checkSlotStatuses] slot leader({}) is not equal with reported data-server({})", slotStats.getSlot().getLeaderDataStoreNodeId(), dataStoreNode.getIp());
                        Metrics.DataSlot.setDataReportNotStable(dataStoreNode.getIp(), slotId);
                        continue;
                    }
                    slotStats.updateLeaderState((LeaderSlotStatus) slotStatus);
                } else {
                    if (!slotStats.getSlot().getFollowerDataStoreNodeIdSet().contains(dataStoreNode.getIp())) {
                        logger.error("[checkSlotStatuses] slot follower({}) is not containing reported data-server({})", slotStats.getSlot().getFollowerDataStoreNodeIdSet(), dataStoreNode.getIp());
                        Metrics.DataSlot.setDataReportNotStable(dataStoreNode.getIp(), slotId);
                        continue;
                    }
                    slotStats.updateFollowerState((FollowerSlotStatus) slotStatus);
                }
            }
        } finally {
            lock.writeLock().unlock();
        }
    }

    @Override
    public void updateSlots(Slots slots) {
        try {
            lock.writeLock().lock();
            logger.info("[updateSlots] update slot table for epoch [{}]", slots.getEpoch());
            slots.getSlotMap().forEach((slotId, slot) -> {
                SlotStats slotStats = slotStatses.get(slotId);
                if (slotStats.getSlot().getLeaderEpoch() < slot.getLeaderEpoch()) {
                    slotStatses.put(slotId, new SlotStatsImpl(slot, serverConfig.getDataReplicateMaxGapMillis()));
                } else
                    if (slotStats.getSlot().getLeaderEpoch() == slot.getLeaderEpoch() && !slotStats.getSlot().equals(slot)) {
                        slotStatses.put(slotId, new SlotStatsImpl(slot, serverConfig.getDataReplicateMaxGapMillis()));
                    } else {
                        logger.warn("[updateSlots]skip slot[{}]", slotId);
                    }
            });
        } finally {
            lock.writeLock().unlock();
        }
    }


}
